package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/13 16:30
 */
public class Flink25_HiveCatalog {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String catalogName = "myhive";  // catalog名字
        String defaultDatabase = "flinktest"; // 数据库名字
        String hiveConfDir = "F:\\atguigu\\01_course\\code\\hive-conf";    // hive配置文件所在目录
//        String hiveConfDir = "/opt/module/hive/conf";    // hive配置文件所在目录

        // TODO 1.创建 HiveCatalog
        HiveCatalog hiveCatalog = new HiveCatalog(catalogName, defaultDatabase, hiveConfDir);
        // TODO 2.注册 HiveCatalog
        tableEnv.registerCatalog(catalogName, hiveCatalog);
        // TODO 3.指定 使用的 catalog
        tableEnv.useCatalog(catalogName);
        // TODO 4.指定 数据库名，不指定就是前面指定的默认的库名
        tableEnv.useDatabase(defaultDatabase);

        // TODO 5.指定 SQL方言
        tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);

        // TODO 5.查询 Hive表
        tableEnv.sqlQuery("select * from test").execute().print();


//        env.execute();
    }
}
